package com.atguigu.app;

import com.atguigu.bean.Bean1;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class LookUpJoinTest1 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //构建主流
        SingleOutputStreamOperator<Bean1> bean1DS = env.socketTextStream("hadoop102", 9999)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new Bean1(fields[0], fields[1], Long.parseLong(fields[2]));
                });
        Table table = tableEnv.fromDataStream(bean1DS,
                $("id"),
                $("name"),
                $("ts"),
                $("pt").proctime());
        tableEnv.createTemporaryView("t1", table);

        //构建维度表
        tableEnv.executeSql("" +
                "create table base_dic( " +
                "    dic_code string, " +
                "    dic_name string, " +
                "    parent_code string, " +
                "    create_time string, " +
                "    operate_time string " +
                ") WITH ( " +
                "  'connector' = 'jdbc', " +
                "  'url' = 'jdbc:mysql://hadoop102:3306/gmall', " +
                "  'table-name' = 'base_dic', " +
//                "  'lookup.cache.max-rows' = '10', " +
//                "  'lookup.cache.ttl' = '1 hour', " +
                "  'username' = 'root', " +
                "  'password' = '123456', " +
                "  'driver' = 'com.mysql.cj.jdbc.Driver' " +
                ")");

        //关联主表与维度表
        tableEnv.sqlQuery("" +
                "select " +
                "    t1.id," +
                "    t1.name," +
                "    dic.dic_name " +
                "from " +
                "    t1 " +
                "join " +
                "    base_dic " +
                "FOR SYSTEM_TIME AS OF t1.pt AS dic " +
                "on t1.id=dic.dic_code")
                .execute()
                .print();

        //tableEnv.sqlQuery("select * from base_dic").execute().print();

    }
}
